package com.aliyun.openservice.iot.as.bridge.demo.core;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.aliyun.iot.as.bridge.core.BridgeBootstrap;
import com.aliyun.iot.as.bridge.core.config.ConfigFactory;
import com.aliyun.iot.as.bridge.core.handler.DownlinkChannelHandler;
import com.aliyun.iot.as.bridge.core.handler.UplinkChannelHandler;
import com.aliyun.iot.as.bridge.core.model.DeviceIdentity;
import com.aliyun.iot.as.bridge.core.model.ProtocolMessage;
import com.aliyun.iot.as.bridge.core.model.Session;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Bridge Gateway Demo for IoT Link Platform Senior Version
 * 物联网平台高级版桥接网关演示
 */
public class BridgeBasicDemo {
	
	// 日志记录器
	private static Logger log = LoggerFactory.getLogger(BridgeBasicDemo.class);

    // 桥接引导程序
    private static BridgeBootstrap bridgeBootstrap;

    /**
     * self-define topic template created in IoT Platform Web Console
     * 在物联网平台Web控制台中创建的自定义主题模板
     */
    private final static String TOPIC_TEMPLATE_USER_DEFINE = "/%s/%s/user/update";
    
    // 发布消息的负载模板
    private final static String PROP_POST_PAYLOAD_TEMPLATE = "Hello IoT Bridge";
    
    // 创建线程池用于处理下行消息
    private static ExecutorService executorService  = new ThreadPoolExecutor(
        Runtime.getRuntime().availableProcessors(),
        Runtime.getRuntime().availableProcessors() * 2,
        60, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(1000),
        new ThreadFactoryBuilder().setDaemon(true).setNameFormat("bridge-downlink-handle-%d").build(),
        new ThreadPoolExecutor.AbortPolicy());

	public static void main(String[] args) throws Exception {
		// 默认使用application.conf和devices.conf配置文件
        bridgeBootstrap = new BridgeBootstrap();
        bridgeBootstrap.bootstrap(new DownlinkChannelHandler() {
            @Override
            public boolean pushToDevice(Session session, String topic, byte[] payload) {
            	// 从云端获取下行消息
                executorService.submit(() -> handleDownLinkMessage(session, topic, payload));
                return true;
            }

            @Override
            public boolean broadcast(String s, byte[] bytes) {
                return false;
            }
        });
        log.info("======== Bridge bootstrap success =========");
        // 原始设备标识，在devices.conf中定义
        String originalIdentity = "12345";

        // 特定协议的设备会话，如果没有，可以使用空对象
        Object originalChannel = new Object();
        Session session = Session.newInstance(originalIdentity, originalChannel);
        // 设备上线
        UplinkChannelHandler uplinkChannelHandler = new UplinkChannelHandler();
        uplinkChannelHandler.doOnline(session, originalIdentity);

        // 发布自定义主题
        // 主题模板${TOPIC_TEMPLATE_USER_DEFINE}在物联网平台Web控制台中定义
        DeviceIdentity deviceIdentity = ConfigFactory.getDeviceConfigManager().getDeviceIdentity(originalIdentity);
        ProtocolMessage protocolMessage = new ProtocolMessage();
        protocolMessage.setPayload(PROP_POST_PAYLOAD_TEMPLATE.getBytes());
        protocolMessage.setQos(0);
        protocolMessage.setTopic(String.format(TOPIC_TEMPLATE_USER_DEFINE, deviceIdentity.getProductKey(), deviceIdentity.getDeviceName()));
        uplinkChannelHandler.doPublishAsync(originalIdentity, protocolMessage);

	}

	// 处理下行消息的方法
	private static void handleDownLinkMessage(Session session, String topic, byte[] payload) {
        String content = new String(payload);
        log.info("Get DownLink message, session:{}, topic:{}, content:{}", session, topic, content);
        Object channel = session.getChannel();
        String originalIdentity = session.getOriginalIdentity();
        // 例如，你可以通过channel将消息发送到设备，这取决于你的具体服务器实现
    }

}
